set -e

MY_DIR=$(basename "$0" | sed 's/\.sh/.d/g')

bin=$(dirname "$0")
bin=$(
		cd "$bin"
		pwd
)

. meta-lib-static/env/.env
echo "[*] DATA_ENGINE_STATIC_DIR: $ACCESS_STATIC_DIR"
DE_STATIC_DIR=$ACCESS_STATIC_DIR

unameOut="$(uname -s)"
case "${unameOut}" in
    Linux*)     OS=linux;;
    Darwin*)    OS=macos;;
esac

mkdir -p ${EXCHANGE_AGENT_PUSH_DATA}.chk

echo "=> data_dir: $EXCHANGE_AGENT_PUSH_DATA"
export EXCHANGE_AGENT_PUSH_DATA EXCHANGE_AGENT_PUSH_KAFKAS EXCHANGE_AGENT_PUSH_PREFIX
export EXCHANGE_AGENT_PUSH_STAGING_AVRO_PORT EXCHANGE_AGENT_PUSH_STAGING_JSON_PORT

SELECTOR=$1
if [ "$1" = "kafka_staging_avro" ]; then
		echo "[*] $DE_STATIC_DIR/$OS/vector-0.12.1 --config $bin/$MY_DIR/buffer_kafka_avro.toml"
		$DE_STATIC_DIR/$OS/vector-0.12.1 --config $bin/$MY_DIR/push_kafka_staging_avro.toml
elif [ "$1" = "kafka_staging_json" ]; then
		echo "[*] $DE_STATIC_DIR/$OS/vector-0.12.1 --config $bin/${MY_DIR}/buffer_kafka_json.toml"
		$DE_STATIC_DIR/$OS/vector-0.12.1 --config $bin/${MY_DIR}/push_kafka_staging_json.toml
elif [ "$1" = "minio" ]; then
		EXCHANGE_AGENT_PUSH_BUCKET=$(echo -ne "$EXCHANGE_AGENT_PUSH_PREFIX" | tr '_' '-')
		if ! $DE_STATIC_DIR/$OS/mc-20210116 alias ls | grep $EXCHANGE_AGENT_PUSH_PREFIX > /dev/null; then
			MNODE=$(echo $EXCHANGE_AGENT_PUSH_MINIO_HOSTS | cut -d, -f1)
			MPORT=$EXCHANGE_AGENT_PUSH_MINIO_PORT
			MNAME=$EXCHANGE_AGENT_PUSH_MINIO_USERNAME
			MPASS=$EXCHANGE_AGENT_PUSH_MINIO_PASSWORD
			$DE_STATIC_DIR/$OS/mc-20210116 alias set $EXCHANGE_AGENT_PUSH_PREFIX http://${MNODE}:${MPORT} $MNAME $MPASS
		fi
		if ! $DE_STATIC_DIR/$OS/mc-20210116 ls $EXCHANGE_AGENT_PUSH_PREFIX/$EXCHANGE_AGENT_PUSH_BUCKET > /dev/null; then
				echo "$DE_STATIC_DIR/$OS/mc-20210116 mb $EXCHANGE_AGENT_PUSH_PREFIX/$EXCHANGE_AGENT_PUSH_BUCKET"
				$DE_STATIC_DIR/$OS/mc-20210116 mb $EXCHANGE_AGENT_PUSH_PREFIX/$EXCHANGE_AGENT_PUSH_BUCKET
		fi
		cat $bin/${MY_DIR}/logrotate.buffer_file.conf | envsubst > /tmp/logrotate.buffer_file.conf
		while true; do
        find ${EXCHANGE_AGENT_PUSH_DATA} -name "*.merge" -delete
				logrotate -s /tmp/logrotate.status.tmp -f /tmp/logrotate.buffer_file.conf
				dataFullPaths=$(find ${EXCHANGE_AGENT_PUSH_DATA} -name "*.old" -exec ls -t "{}" +)
        for dataFullPath in $dataFullPaths; do
					dataSubPath=$(echo $dataFullPath | sed "s|${EXCHANGE_AGENT_PUSH_DATA}/||g")
					
					file_dir=$(dirname $dataSubPath)
					file_name=$(basename $dataSubPath) # 10.25.21.247_3029_20.staging_json.txt.1626838259.old
					node_str=$(echo $file_name | cut -d_ -f1,2)
					type_str=$(echo $file_name | cut -d. -f5)
					ts_str=$(date '+%Y%m%d%H%M%S' -d @$(echo $file_name | cut -d. -f7))
					dt_str=$(echo $ts_str | cut -b1-8)
					tm_str=$(echo $ts_str | cut -b9-12)
					
					dataSubPathNew=${file_dir}/dt=${dt_str}/${node_str}_${tm_str}.${type_str}.txt
          mkdir -p $(dirname ${EXCHANGE_AGENT_PUSH_DATA}/${dataSubPathNew}.merge)
					cat ${EXCHANGE_AGENT_PUSH_DATA}/${dataSubPath} | cut -d'|' -f2- >> ${EXCHANGE_AGENT_PUSH_DATA}/${dataSubPathNew}.merge
				done

				dataMergeFullPaths=$(find ${EXCHANGE_AGENT_PUSH_DATA} -name "*.merge" -exec ls -t "{}" +)
				for dataMergeFullPath in $dataMergeFullPaths; do
					dataSubPath=$(echo $dataMergeFullPath | sed "s|${EXCHANGE_AGENT_PUSH_DATA}/||g")
          if [ -s $dataMergeFullPath ]; then
					  while true; do
					  	$DE_STATIC_DIR/$OS/mc-20210116 cp --disable-multipart ${EXCHANGE_AGENT_PUSH_DATA}/${dataSubPath} $EXCHANGE_AGENT_PUSH_PREFIX/$EXCHANGE_AGENT_PUSH_BUCKET/${dataSubPath}
					  	if [ $? -eq 0 ]; then break; fi
					    echo " [WARN]: COPY TO MINIO FAILED! RETRY AFTER 2 SECONDS..."
					    sleep 2
					  done
          fi
					rm -rf ${EXCHANGE_AGENT_PUSH_DATA}/${dataSubPath}
        done
				echo sleep $(( 60 * $EXCHANGE_AGENT_PUSH_ROLL_MINUTES ))
				sleep $(( 60 * $EXCHANGE_AGENT_PUSH_ROLL_MINUTES ))
        find ${EXCHANGE_AGENT_PUSH_DATA} -name "*.old" -delete
		done
fi
